package com.navi.hms;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.navi.FileLoader;
import com.navi.util.JacksonUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;
import org.apache.commons.lang.StringUtils;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
import org.springframework.test.context.junit4.SpringRunner;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * Created by lin.xin on 2017/9/21.
 */
@SpringBootTest
@RunWith(SpringRunner.class)
public class BuildCsvFileFromRabbitMQJob {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${gp.datasource.drive}")
    private String driver;
    @Value("${gp.datasource.url}")
    private String url;
    @Value("${gp.datasource.username}")
    private String username;
    @Value("${gp.datasource.password}")
    private String password;
    @Value("${spring.rabbitmq.queues}")
    private String queues;
    @Value("${spring.rabbitmq.exchange}")
    private String exchange;
    @Value("${local.csv.path}")
    private String localCsvPath;


    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private FileLoader fileLoader;

    @Bean
    public TaskScheduler taskScheduler() {
        return new ConcurrentTaskScheduler(); //single threaded by default
    }



    public void buildCsvFile() {
        logger.info("hms system start ......");
        List<JSONObject> jsonObjectList = new ArrayList<>();

        Arrays.stream("HMS_ERROR".split(",")).parallel().forEach(queue -> {
            try (Connection connection = connectionFactory.newConnection()) {
                logger.info("connected to RabbitMQ server");
                Channel channel = connection.createChannel();
                channel.exchangeDeclare(exchange, ExchangeTypes.TOPIC, true);
                logger.info("connected to RabbitMQ exchange  {} ", exchange);
                while (true) {
                    final GetResponse response1 = channel.basicGet(queue, false);
                    if (response1 == null) {
                        break;
                    }
                    final byte[] body1 = response1.getBody();
                    String message = new String(body1, "utf-8");
                    if (StringUtils.startsWith(message, "{")) {
                        jsonObjectList.add((JSONObject) JSONObject.parse(message));
                    } else if (StringUtils.startsWith(message, "[")) {
                        jsonObjectList.addAll(JSONArray.parseArray(message).stream().map(obj -> (JSONObject) obj).collect(Collectors.toList()));
                    }
                    //确认已消费成功，返回ACK确认
                    channel.basicAck(response1.getEnvelope().getDeliveryTag(), true);
                }
            } catch (Exception ex) {
                ex.printStackTrace();
                logger.error(ex.toString());
            }
        });
        Map<String, List<JSONObject>> tableJsonMap = jsonObjectList.stream().collect(Collectors.groupingBy(json -> json.getString("ope_tbl_name")));
//        Map<String, String> tableFileMap = new HashMap<>();
        tableJsonMap.forEach((tableName, jsonList) -> {
            //get table's column
            try (Connection connection = connectionFactory.newConnection()){
                Channel channel = connection.createChannel();
                channel.exchangeDeclare(exchange, ExchangeTypes.TOPIC, true);
               jsonList.forEach(jsonObject -> {
                   if ("".equalsIgnoreCase(jsonObject.getString(""))) {
                       try {
                           channel.basicPublish("", "HMS_ERROR2", null, JacksonUtil.toJSONStr(jsonObject).getBytes("UTF-8"));
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   } else {
                       try {
                           channel.basicPublish("", "HMS_ERROR", null, JacksonUtil.toJSONStr(jsonObject).getBytes("UTF-8"));
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }

               });
            } catch (Exception e) {
                e.printStackTrace();
                logger.error(e.toString());
            }
        });
    }

    public static String underline2Camel(String line, boolean smallCamel) {
        if (line == null || "".equals(line)) {
            return "";
        }
        StringBuffer sb = new StringBuffer();
        Pattern pattern = Pattern.compile("([A-Za-z\\d]+)(_)?");
        Matcher matcher = pattern.matcher(line);
        while (matcher.find()) {
            String word = matcher.group();
            sb.append(smallCamel && matcher.start() == 0 ? Character.toLowerCase(word.charAt(0)) : Character.toUpperCase(word.charAt(0)));
            int index = word.lastIndexOf('_');
            if (index > 0) {
                sb.append(word.substring(1, index).toLowerCase());
            } else {
                sb.append(word.substring(1).toLowerCase());
            }
        }
        return sb.toString();
    }
}
